using System;
using System.IO;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using Microsoft.BizTalk.TransportProxy.Interop;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.XLANGs.BaseTypes;

namespace Microsoft.Samples.BizTalk.Adapter.Tcp.Send
{
    /// <summary>
    ///     This class represents a worker thread that is used to submit batches to BizTalk.
    /// </summary>
    /// <author>
    ///     Paul Brown
    /// </author>
    public class BatchWorkerThread : WorkerThread
    {
        #region Private Instance Members

        private SendAdapter                             _sendAdapter;       // The send adapter

        #endregion

        #region Constructor

        /// <summary>
        ///     This constructor initializes the instance.
        /// </summary>
        public BatchWorkerThread(IAdapter adapter)
            : base(adapter)
        {
            // Get thread name prefix
            this.ThreadNamePrefix = TraceHelper.ResourceManager.GetString(ResourceConstants.SendBatchWorkerThreadName);

            // We know the adapter will be a send adapter, store a reference to it
            this._sendAdapter = (SendAdapter)adapter;
        }

        #endregion

        #region Public Instance Properties

        /// <summary>
        ///     This property returns a reference to the send adapter.
        /// </summary>
        public SendAdapter SendAdapter
        {
            get { return (this._sendAdapter); }
        }

        #endregion

        #region Private Instance Methods

        #endregion

        #region Abstract WorkerThread Members

        #region Public Instance Properties

        /// <summary>
        ///     This property returns the type of a BatchWorkItem which is compatible with this worker thread.
        /// </summary>
        public override Type WorkItemType
        {
            get { return (typeof(BatchWorkItem)); }
        }

        #endregion

        #region Protected Instance Methods

        /// <summary>
        ///     This method is the delegate that runs the main worker thread for unordered batches.
        /// </summary>
        protected override void Thread()
        {
            // Variables used to control batching
            bool submitted = false;
            bool        threadInterrupted = false;
            int             messageCount;
            long            batchDataSize;
            Stopwatch       timer = new Stopwatch();
            BatchWorkItem   workItem = null;

            try
            {
                // Loop until we are interrupted
                while (true)
                {
                    // If we have work, start batching process
                    if (this.CurrentWorkLoad > 0)
                    {
                        if (this.Adapter.TraceSwitch.TraceVerbose)
                            TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendGetBatch, this.Name);

                        // Create a new batch for BizTalk
                        Batch batch = new Batch(this.Adapter);
                        batch.RetryCount = this.SendAdapter.BatchRetryCount;
                        batch.RetryInterval = this.SendAdapter.BatchRetryInterval;
                        batch.OrderedBatch = false;

                        // Reset batch controlling properties
                        messageCount = 0;
                        batchDataSize = 0;
                        timer.Reset();
                        timer.Start();

                        // Loop while we have work items
                        while (this.CurrentWorkLoad > 0)
                        {
                            if (submitted == true)
                            {
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ReceiveInitialiseNewBatch, this.Name);
                                batch = new Batch(this.Adapter);
                                batch.RetryCount = this.SendAdapter.BatchRetryCount;
                                batch.RetryInterval = this.SendAdapter.BatchRetryInterval;
                                batch.OrderedBatch = false;

                                // Reset batch controlling properties
                                messageCount = 0;
                                batchDataSize = 0;
                                timer.Reset();
                                timer.Start();
                                submitted = false;
                            }
                            
                            // Dequeue item
                            workItem = (BatchWorkItem)this.DequeueWork();

                            if (this.Adapter.TraceSwitch.TraceVerbose)
                                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendDequeuedWork, workItem.SendLocation.Uri, this.Name);

                            // Set its status
                            workItem.Status = WorkItemStatusTypes.InProgress;

                            // Add batch messages in work item to batch
                            long size = 0;
                            foreach (BatchMessage message in workItem.BatchMessages)
                            {
                                bool addedSuccessfully = false;
                                while (!addedSuccessfully)
                                {
                                    try
                                    {
                                        // Add to batch (only certain batch operations need supporting in a send adapter)
                                        switch (message.BatchOperationType)
                                        {
                                            case BatchOperationType.Submit :

                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendAddingResponseToBatch, this.Name, workItem.SendLocation.Uri);

                                                batch.SubmitResponseMessage(message);
                                                // [JPLUM: Codeplex Workitem: 6548]
                                                if (message.Message.BodyPart != null)
                                                {
                                                    // 6549 size = message.Message.GetPart("body").Data.Length; 
                                                    size = message.Message.BodyPart.Data.Length;
                                                }
                                                else
                                                {
                                                    size = 0;
                                                }
                                                break;
                                                
                                            case BatchOperationType.Delete :

                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendAddingDeleteToBatch, this.Name, workItem.SendLocation.Uri);

                                                batch.DeleteMessage(message);
                                                break;

                                            case BatchOperationType.Resubmit :

                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendAddingResubmitToBatch, this.Name, workItem.SendLocation.Uri);

                                                batch.ResubmitMessage(message);
                                                break;

                                            case BatchOperationType.MoveToNextTransport:

                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendAddingMoveToNextTransportToBatch, this.Name, workItem.SendLocation.Uri);

                                                batch.MoveToBackupTransport(message);
                                                break;

                                            case BatchOperationType.MoveToSuspendQ:

                                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendAddingMoveToSuspendQueueToBatch, this.Name, workItem.SendLocation.Uri);

                                                batch.MoveToSuspendQ(message);
                                                
                                            // [JPLUM: Codeplex Workitem: 6548]
                                                if (message.Message.BodyPart != null)
                                                {
                                                    // 6549 size = message.Message.GetPart("body").Data.Length; 
                                                    size = message.Message.BodyPart.Data.Length;
                                                }
                                                else
                                                {
                                                    size = 0;
                                                }
                                                break;

                                        }

                                        addedSuccessfully = true;
                                    }
                                    catch (ThreadInterruptedException)
                                    {
                                        // Asked to shut down, set flag and wait until this batch is done
                                        threadInterrupted = true;
                                    }
                                    catch (Exception e)
                                    {
                                        TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.SendFailedAddingToBatch, workItem.SendLocation.Uri, batch.BatchCookie, e.Message);
                                        addedSuccessfully = false;
                                    }
                                }

                                // Increment controlling properties
                                messageCount++;
                                batchDataSize += size;
                            }

                            // Anything in batch?
                            if (messageCount > 0)
                            {
                                bool readyToSubmit = false;

                                // Have we exceeded our batch limits?
                                if (messageCount >= this.SendAdapter.ResponseBatchSize)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendReachedBatchSizeLimit, this.SendAdapter.ResponseBatchSize, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (batchDataSize >= this.SendAdapter.MaxBatchData)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendExceededMaxBatchDataLimit, this.SendAdapter.MaxBatchData, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (timer.ElapsedMilliseconds >= this.SendAdapter.BatchTimeout)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendExceededBatchTimeoutLimit, this.SendAdapter.BatchTimeout, messageCount);

                                    readyToSubmit = true;
                                }
                                else if (this.CurrentWorkLoad == 0)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendSubmittingBatch, messageCount);

                                    readyToSubmit = true;
                                }

                                // Are we ready to submit?
                                if (readyToSubmit)
                                {
                                    // Add to array
                                    if (Batch.GetBatch(batch.BatchCookie) == null)
                                    {
                                        Batch.AddBatch(batch);
                                    }

                                    // Submit batch if limits exceeded or idle
                                  //  bool submitted = false;
                                    for (int i = 0; i < this.SendAdapter.BatchRetryCount; i++)
                                    {
                                        try
                                        {
                                            // Submit batch
                                            batch.SubmitBatch();

                                            // Break out of loop, successful
                                            submitted = true;
                                            break;
                                        }
                                        catch (ThreadInterruptedException)
                                        {
                                            // Asked to shut down, set flag and wait until this batch is done
                                            threadInterrupted = true;
                                        }
                                        catch (Exception e)
                                        {
                                            // Trace
                                            TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.SendFailedSubmittingBatchRetrying, batch.BatchCookie, e.Message);

                                            // Wait interval
                                            System.Threading.Thread.Sleep(this.SendAdapter.BatchRetryInterval * 1000);
                                        }
                                    }

                                    // If not successful and exhausted all retries
                                    if (!submitted)
                                    {
                                        TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.SendFailedSubmittingBatch, batch.BatchCookie);

                                        // Save data
                                        workItem.Save();
                                    }
                                }
                            }
                            else
                            {
                                // No messages in batch!
                                if (this.Adapter.TraceSwitch.TraceVerbose)
                                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.SendNoMessagesInBatch);
                            }

                            // Update work item
                            workItem.Status = WorkItemStatusTypes.Complete;
                            workItem = null;
                        }
                    }

                    // Stop timer
                    timer.Stop();

                    // Was the thread interrupted?
                    if (threadInterrupted)
                    {
                        // Throw error again to end thread
                        throw (new ThreadInterruptedException());
                    }

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadWaitingForWork, this.Name);

                    // Wait for signal that work has arrived
                    this.WaitForWork();
                }
            }
            catch (ThreadInterruptedException)
            {
                // Asked to shut down
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.WorkerThreadInterrupted, this.Name);

                // If we have any in-flight work still, save it off
                if (workItem != null)
                {
                    workItem.Save();
                }

                while (this.CurrentWorkLoad > 0)
                {
                    workItem = (BatchWorkItem)this.DequeueWork();
                    workItem.Save();
                }
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.WorkerThreadGeneralError, this.Name, e.Message);

                // If we have an outstanding work item, push it back onto queue before restarting
                if (workItem != null)
                {
                    this.PushWorkOnQueue(workItem);
                }
            }
            finally
            {
                // Fire terminate event
                this.OnThreadTerminated(new EventArgs());
            }
        }

        #endregion

        #endregion
    }
}
